package com.kevin.eshopcache.kafka;

import com.alibaba.fastjson.JSONObject;
import com.kevin.eshopcache.entity.ProductInfo;
import com.kevin.eshopcache.entity.ShopInfo;
import com.kevin.eshopcache.service.CacheService;
import com.kevin.eshopcache.utils.ContextUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Optional;

/**
 * @author: e20160504
 * @date: 2018/11/9
 * @desc
 */
public class Process implements Runnable {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private ConsumerRecord<?, ?> record;
    private CacheService cacheService;

    public Process(ConsumerRecord<?, ?> record) {
        this.record = record;
        this.cacheService = (CacheService) ContextUtil.getApplicationContext().getBean("cacheService");
    }

    @Override
    public void run() {

        Optional<?> messages = Optional.ofNullable(record.value());
        if (messages.isPresent()) {
            String msg = (String) messages.get();

            JSONObject msgJsonObject = JSONObject.parseObject(msg);
            String serviceId = msgJsonObject.getString("serviceId");
            //如果是商品信息服务
            if ("productInfoService".equals(serviceId)) {
                processProductInfoMessage(msgJsonObject);
            } else if ("shopInfoService".equals(serviceId)) {
                processShopInfoMessage(msgJsonObject);
            }
        }
    }


    private void processProductInfoMessage(JSONObject msgJsonObject) {

        Long Objid = msgJsonObject.getLong("id");

        //调用商品信息服务接口，将productId作为参数传递过去，一般是从数据库中取，然后返回结果，结果一般为json
        String productInfoJson = "{\"id\":1,\"name\":\"iphone8\",\"price\":588.88}";
        ProductInfo productInfo = JSONObject.parseObject(productInfoJson, ProductInfo.class);

        cacheService.saveLocalCache(productInfo);
        logger.info("====获取本地缓存的商品信息 {}",cacheService.getLocalCache(Objid));
        cacheService.saveRedisProductCache(productInfo);
    }

    private void processShopInfoMessage(JSONObject msgJsonObject) {
        //取出店铺Id
        Long Objid = msgJsonObject.getLong("id");

        //调用商品信息服务接口，将productId作为参数传递过去，一般是从数据库中取，然后返回结果，结果一般为json
        String shopInfoJson = "{\"id\":1,\"name\":\"Kevin shop\"}";
        ShopInfo shopInfo = JSONObject.parseObject(shopInfoJson, ShopInfo.class);

        cacheService.saveLocalShopCache(shopInfo);
        logger.info("====获取本地缓存的商品信息 {}",cacheService.getLocalShopCache(Objid));
        cacheService.saveRedisShopCache(shopInfo);
    }
}
